package com.eric.demo.kafka;

import com.codahale.metrics.Counter;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.IOException;
import java.util.Date;
import java.util.List;

/**
 * @author ZhangGuangyin on 2018/8/9
 */
public class Consumer implements Runnable{

    KafkaConsumer<String, String> kafkaConsumer;

    List<String> topicList;

    ObjectMapper mapper = new ObjectMapper();

    Counter counter;

    public Consumer(List<String> topicList, Counter counter){
        kafkaConsumer = new KafkaConsumer<>(ConsumerProperties.PROPERTIES.getProperties());
        this.topicList = topicList;
        kafkaConsumer.subscribe(topicList);
        this.counter = counter;
        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

    @Override
    public void run() {
        while (true){
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Constants.POLL_TIME_OUT);
            if (records != null && !records.isEmpty()) {
                for (ConsumerRecord<String, String> record : records) {
                    if (record != null) {
                        counter.inc();
                        String message = record.value();
                        Date receiveTime = new Date();
                        try {
                            Wares wares = mapper.readValue(message, Wares.class);
                            wares.setReceiveTime(receiveTime);
                        } catch (IOException e) {
                            System.out.println("err:" + message);
                        }
                    }
                }

            }
        }
    }
}
